package com.to8to.kitt.asyc;

import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.ReferenceCountUtil;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.thrift.protocol.TMessageMeta;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.to8to.kitt.ThriftMessage;
import com.to8to.kitt.utils.LogUtils;
import com.to8to.kitt.utils.NettyUtils;

@Sharable
public class AsyncBizChannelHandler extends ChannelHandlerAdapter {
	
	public static Logger logger = LoggerFactory.getLogger(AsyncBizChannelHandler.class);  

	private Map<Integer, State> states = new ConcurrentHashMap<Integer, State>();
	
	private Map<String, IMsgHandler> name2handlers = new ConcurrentHashMap<String, IMsgHandler>();
	
	private IMsgHandler defaultHandler;
	
	private int msgTimeoutMs = Integer.MAX_VALUE;
	
	private ExecutorService executor = Executors.newCachedThreadPool();
	
	public AsyncBizChannelHandler(Map<String, IMsgHandler> handlers, IMsgHandler defaultHandler)
	{
		this.name2handlers.putAll(handlers);
		
		this.defaultHandler = defaultHandler;
		
		lanuchCleanerThread();
	}
	
	public AsyncBizChannelHandler(Map<String, IMsgHandler> handlers)
	{
		this.name2handlers.putAll(handlers);
		
		lanuchCleanerThread();
	}
	
	public AsyncBizChannelHandler(IMsgHandler defaultHandler)
	{
		this.defaultHandler = defaultHandler;
	}
	
	
	public void lanuchCleanerThread()
	{
		Thread t = new Thread(new Runnable(){

			@Override
			public void run() 
			{	
				while(true)
				{
					long now = System.currentTimeMillis();
					
					Set<Integer> removeSeq = new HashSet<Integer>();
					
					for(Integer key: states.keySet())
					{
						State status = states.get(key);
						
						if(now - status.getTime() > msgTimeoutMs)
							removeSeq.add(key);
					}
					
					for(Integer key: removeSeq)
						states.remove(key);
					
					try {
						Thread.sleep(msgTimeoutMs);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
				
			}});
		
		t.start();
	}
	
	public Map<String, IMsgHandler> getName2handlers() {
		return name2handlers;
	}

	public void setName2handlers(Map<String, IMsgHandler> name2handlers) {
		this.name2handlers = name2handlers;
	}

	public IMsgHandler getDefaultHandler() {
		return defaultHandler;
	}

	public void setDefaultHandler(IMsgHandler defaultHandler) {
		this.defaultHandler = defaultHandler;
	}


	
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }
    

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) 
    {
    	
    	ThriftMessage tmsg = (ThriftMessage)msg;
    	
    	LogUtils.openTag(tmsg.getMeta().tag);
    	
    	logger.debug("===> 收到客户端请求! 对方地址:{} 消息头:{}", ctx.channel().remoteAddress(), tmsg.getHead());
    	
    	logger.trace("======> 请求包体:{}", tmsg.getBody());
    	
    	executor.execute(new Runnable(){

			@Override
			public void run() {
				try {
					handle(ctx, msg);
		        } 
		    	catch(Throwable e)
		    	{
		    		NettyUtils.exitIfOutOfMemory(e);
		    		
		    		logger.error("Exception", LogUtils.logStackTrace(e));
		    	}
		    	finally {
		        	ReferenceCountUtil.release(msg);
		        }
			}});
    	
    	LogUtils.removeTag();
    }
    
    
	public void handle(ChannelHandlerContext ctx, Object msg) 
	{
		ThriftMessage inMsg = (ThriftMessage)msg;
		
		TMessageMeta meta = inMsg.getMeta();
		
		LogUtils.openTag(meta.tag);
		
		logger.debug("开始业务处理:{} ", inMsg.getHead().name);
		
		IMsgHandler handler =  null;
		
		if(inMsg.isCall())
		{
			String serviceMethod = meta.serviceMethod();
			
			handler = name2handlers.get(serviceMethod);
			
			if(handler == null)
			{
				logger.debug("找不到handler: {}, 因此用默认的handler", serviceMethod);
				handler = defaultHandler;
			}
		}
		else 
		{
			State state = states.get(inMsg.getSeqid());
			if(state != null)
				handler = state.getHandler();
			else 
				logger.error("找不到序列号对应的状态: " + inMsg.getSeqid());
		}

		if(handler != null)
			handler.handle(ctx, inMsg, states);
		else 
		{
			logger.debug("找不到handler: {}", inMsg.getHead().name);
		}
		
		logger.debug("完成业务处理:{} ", inMsg.getHead().name);
		
		LogUtils.removeTag();
	}

	
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        NettyUtils.exitIfOutOfMemory(cause);
    }
}
